package com.zhihaoliang.coroutine.buss.net.data

import com.zhihaoliang.coroutine.buss.net.callback.IUserLoadData
import com.zhihaoliang.coroutine.buss.net.model.Repo
import com.zhihaoliang.coroutine.buss.net.model.RequestData
import com.zhihaoliang.coroutine.buss.net.model.User
import com.zhihaoliang.coroutine.buss.net.util.http.GitHubService
import com.zhihaoliang.coroutine.buss.net.util.http.NetCallback
import com.zhihaoliang.coroutine.buss.net.util.http.createGitHubService
import com.zhihaoliang.coroutine.util.aggregate
import com.zhihaoliang.coroutine.util.log
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.Channel
import java.util.concurrent.atomic.AtomicInteger

/**
 * @author 支豪亮
 * @date 2022/10/13 17:50
 * @des 网络数据
 */
class NetworkDataSource {
    /**
     *  github 账号的设置
     *  登录账号、token 、组织
     *  获取token的地址 https://github.com/settings/tokens/new
     */
    private val req = RequestData("zhihaoliang07@163.com",
                                  "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", "kotlin")

    private val service: GitHubService = createGitHubService(req.username, req.password)


    /**
     * 线程 execute 阻塞
     */
    fun loadContributorsBlocking(): List<User> {
        val repos = service.getOrgReposCall(req.org).execute().body() ?: emptyList()

        return repos.flatMap {
            service.getRepoContributorsCall(req.org, it.name).execute().body() ?: emptyList()
        }.aggregate()
    }

    /**
     * 异步 enqueue 回调
     */
    fun loadContributorsCallbacks(userLoadData: IUserLoadData) {
        service.getOrgReposCall(req.org).enqueue(NetCallback {
            val listRepos: List<Repo> = it ?: emptyList()
            if (listRepos.isEmpty()) userLoadData.onUserLoad(emptyList())

            val allUsers: MutableList<User> = ArrayList()

            val numberOfProcessed = AtomicInteger() //多线程的同步处理 //val countDownLatch = CountDownLatch(listRepos.size)

            for (repo in listRepos) {
                service.getRepoContributorsCall(req.org, repo.name).enqueue(NetCallback { list ->

                    if (!list.isNullOrEmpty()) allUsers.addAll(list)

                    if (numberOfProcessed.incrementAndGet() == listRepos.size) {
                        userLoadData.onUserLoad(allUsers.aggregate())
                    }
                })
            } //countDownLatch.await()  //导致线程阻塞
            // userLoadData.onUserLoad(allUsers.aggregate())
        })
    }

    /**
     * 协成加载
     */
    suspend fun loadContributorsSuspend(userLoadData: IUserLoadData) {
        val repos = service.getOrgRepos(req.org).body() ?: emptyList()

        val users = repos.flatMap { repo ->
            service.getRepoContributors(req.org, repo.name).body() ?: emptyList()
        }.aggregate()

        userLoadData.onUserLoad(users)
    }

    /**
     *  协成并行
     */
    suspend fun loadContributorsConcurrent(userLoadData: IUserLoadData) {
        val repos = service.getOrgRepos(req.org).body() ?: emptyList()

        val deferred = repos.map { repo ->
            CoroutineScope(Dispatchers.IO).async {
                log(repo.name, coroutineContext)
                service.getRepoContributors(req.org, repo.name).body() ?: emptyList()
            }
        }

        val list = deferred.awaitAll().flatten().aggregate()

        userLoadData.onUserLoad(list)

    }

    /**
     * 协成加载进度
     */
    suspend fun loadContributorsProgress(userLoadData: IUserLoadData) {
        val repos = service.getOrgRepos(req.org).body() ?: emptyList()
        var allUsers = emptyList<User>()
        for (repo in repos) {
            val list = service.getRepoContributors(req.org, repo.name).body() ?: emptyList()
            allUsers = (allUsers + list).aggregate()
            userLoadData.onUserLoad(allUsers)
        }
    }

    /**
     * 协成通道
     */
    suspend fun loadContributorsChannels(userLoadData: IUserLoadData) = coroutineScope {

        val repos = service.getOrgRepos(req.org).body() ?: emptyList()

        val channel = Channel<List<User>>(50)
        for (repo in repos) {
            launch(Dispatchers.IO) {
                val users = service.getRepoContributors(req.org, repo.name).body() ?: emptyList()
                channel.send(users)
            }
        }
        var allUsers = emptyList<User>()
        repeat(repos.size) {
            val users = channel.receive()
            allUsers = (allUsers + users).aggregate()
            userLoadData.onUserLoad(allUsers)
        }
    }

}